package com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.controller;

import com.qupeng.demo.kafka.kafkaspringbootstreams.demo.CustomerComplaint;
import com.qupeng.demo.kafka.kafkaspringbootstreams.common.controller.CommonResponse;
import com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.utils.KafkaService;
import com.qupeng.demo.kafka.kafkaspringbootstreams.common.controller.ProduceSuccessResponse;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentCase;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentPresence;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentProfile;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.*;

import java.util.Map;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/message")
public class DashboardController {
    @Autowired
    private KafkaService<String, AgentProfile> profileService;

    @Autowired
    private KafkaService<String, AgentCase> caseService;

    @Autowired
    private KafkaService<String, AgentPresence> presenceService;

    @Autowired
    private KafkaService<String, CustomerComplaint> complaintService;

    @PostMapping("/agent-profiles")
    public CommonResponse<ProduceSuccessResponse<String, AgentProfile>> produceProfile(@RequestBody AgentProfile agentProfile) {
        return this.<String, AgentProfile>produce(profileService, agentProfile, "agent-profiles", agentProfile.getId());
    }
    @PostMapping("/agent-cases")
    public CommonResponse<ProduceSuccessResponse<String, AgentCase>> produceCase(@RequestBody AgentCase agentCase) {
        return this.<String, AgentCase>produce(caseService, agentCase, "agent-cases", agentCase.getAgentId());
    }

    @PostMapping("/agent-presences")
    public CommonResponse<ProduceSuccessResponse<String, AgentPresence>> produceCase(@RequestBody AgentPresence agentPresence) {
        return this.<String, AgentPresence>produce(presenceService, agentPresence, "agent-presences", agentPresence.getAgentId());
    }

    @PostMapping("/agent/data")
    public CommonResponse<ProduceSuccessResponse<String, String>> producePeter(@RequestBody Map<String, String> requestBody) {
        String agentIndex = requestBody.get("agentIndex");
        AgentProfile agentProfile = new AgentProfile("u00" + agentIndex, requestBody.get("agentName"), 5);
        this.<String, AgentProfile>produce(profileService, agentProfile, "agent-profiles", agentProfile.getId());
        produce(requestBody.get("dataIndex"), agentIndex);

        return new CommonResponse<>("producer");
    }

    private void produce(String dataIndex, String agentIndex) {
        for (int i = 0; i < 10; i++) {
            AgentCase agentCaseHandling = new AgentCase("c" + dataIndex + "0" + i, "u00" + agentIndex, "HANDLING");
            this.<String, AgentCase>produce(caseService, agentCaseHandling, "agent-cases", agentCaseHandling.getAgentId());
            AgentCase agentCaseHandled = new AgentCase("c" + dataIndex + "0" + ++i, "u00" + agentIndex, "CLOSED");
            this.<String, AgentCase>produce(caseService, agentCaseHandled, "agent-cases", agentCaseHandled.getAgentId());
            AgentPresence agentPresence = new AgentPresence("p" + dataIndex + "0"  + i, "u00" + agentIndex, "ONLINE", 10);
            this.<String, AgentPresence>produce(presenceService, agentPresence, "agent-presences", agentPresence.getAgentId());
        }
    }

    private <K, V>CommonResponse<ProduceSuccessResponse<K, V>> produce(KafkaService<K, V> kafkaService, V messagePayload, String topic, K key) {
        try {
            SendResult<K, V> sendResult = kafkaService.sendToKafkaSync(topic, key, messagePayload);
            return new CommonResponse<>("producer", new ProduceSuccessResponse<K, V>(sendResult));
        } catch (ExecutionException | InterruptedException e) {
            return new CommonResponse<>("001-producer-controller", e.getMessage(), "producer");
        }
    }
}
